﻿using System.Collections.Generic;
using System.Diagnostics.Contracts;
#if !EXPERIMENTAL
using System.Linq;
#endif
using System.Reactive.Disposables;

namespace System.Reactive.Linq
{
	public static partial class Observable2
	{
#if !EXPERIMENTAL
#if !WINDOWS_PHONE
		/// <summary>
		/// Merges two observable sequences into one observable sequence of tuples by combining elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining pairwise elements from each source into a tuple.</returns>
		public static IObservable<Tuple<T1, T2>> Zip<T1, T2>(this IObservable<T1> first, IObservable<T2> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2>>>() != null);

			return first.Zip(second, Tuple.Create);
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence of tuples by combining elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining pairwise elements from each source into a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3>> Zip<T1, T2, T3>(this IObservable<Tuple<T1, T2>> first, IObservable<T3> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3>>>() != null);

			return first.Zip(second, (t, t3) => Tuple.Create(t.Item1, t.Item2, t3));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence of tuples by combining elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining pairwise elements from each source into a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4>> Zip<T1, T2, T3, T4>(this IObservable<Tuple<T1, T2, T3>> first, IObservable<T4> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4>>>() != null);

			return first.Zip(second, (t, t4) => Tuple.Create(t.Item1, t.Item2, t.Item3, t4));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence of tuples by combining elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <typeparam name="T5">The type of the tuple's fifth component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining pairwise elements from each source into a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4, T5>> Zip<T1, T2, T3, T4, T5>(this IObservable<Tuple<T1, T2, T3, T4>> first, IObservable<T5> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4, T5>>>() != null);

			return first.Zip(second, (t, t5) => Tuple.Create(t.Item1, t.Item2, t.Item3, t.Item4, t5));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence of tuples by combining elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <typeparam name="T5">The type of the tuple's fifth component.</typeparam>
		/// <typeparam name="T6">The type of the tuple's sixth component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining pairwise elements from each source into a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4, T5, T6>> Zip<T1, T2, T3, T4, T5, T6>(this IObservable<Tuple<T1, T2, T3, T4, T5>> first, IObservable<T6> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4, T5, T6>>>() != null);

			return first.Zip(second, (t, t6) => Tuple.Create(t.Item1, t.Item2, t.Item3, t.Item4, t.Item5, t6));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence of tuples by combining elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <typeparam name="T5">The type of the tuple's fifth component.</typeparam>
		/// <typeparam name="T6">The type of the tuple's sixth component.</typeparam>
		/// <typeparam name="T7">The type of the tuple's seventh component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining pairwise elements from each source into a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7>> Zip<T1, T2, T3, T4, T5, T6, T7>(this IObservable<Tuple<T1, T2, T3, T4, T5, T6>> first, IObservable<T7> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7>>>() != null);

			return first.Zip(second, (t, t7) => Tuple.Create(t.Item1, t.Item2, t.Item3, t.Item4, t.Item5, t.Item6, t7));
		}

		/// <summary>
		/// Merges two observable sequences into one observable sequence of tuples by combining elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="T1">The type of the tuple's first component.</typeparam>
		/// <typeparam name="T2">The type of the tuple's second component.</typeparam>
		/// <typeparam name="T3">The type of the tuple's third component.</typeparam>
		/// <typeparam name="T4">The type of the tuple's fourth component.</typeparam>
		/// <typeparam name="T5">The type of the tuple's fifth component.</typeparam>
		/// <typeparam name="T6">The type of the tuple's sixth component.</typeparam>
		/// <typeparam name="T7">The type of the tuple's seventh component.</typeparam>
		/// <typeparam name="T8">The type of the tuple's eighth component.</typeparam>
		/// <param name="first">The first observable sequence.</param>
		/// <param name="second">The second observable sequence.</param>
		/// <returns>An observable sequence containing the result of combining pairwise elements from each source into a tuple.</returns>
		public static IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7, T8>> Zip<T1, T2, T3, T4, T5, T6, T7, T8>(this IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7>> first, IObservable<T8> second)
		{
			Contract.Requires(first != null);
			Contract.Requires(second != null);
			Contract.Ensures(Contract.Result<IObservable<Tuple<T1, T2, T3, T4, T5, T6, T7, T8>>>() != null);

			return first.Zip(second, (t, t8) => new Tuple<T1, T2, T3, T4, T5, T6, T7, T8>(t.Item1, t.Item2, t.Item3, t.Item4, t.Item5, t.Item6, t.Item7, t8));
		}
#endif

		/// <summary>
		/// Merges two or more observable sequences into one observable sequence of lists by combining their elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="TSource">The object that provides notification information.</typeparam>
		/// <param name="sources">An enumerable sequence containing two or more observable sequences to be merged.</param>
		/// <remarks>
		/// <para>
		/// All of the observable sequences must produce at least one element; otherwise, the generated sequence will be empty.  Furthermore, if an observable 
		/// sequence produces more than one element before each of the other observable sequences have produced their first elements, then all of the elements 
		/// will be enqueued for future pairing.  No elements are discarded.
		/// </para>
		/// <para>
		/// The latest value of an observable sequence is always located in the generated lists at the same index in which that sequence is located in the outer sequence.
		/// For example, the values from the first observable sequence in the outer sequence will always be at index zero (0) in the lists that are generated.
		/// </para>
		/// </remarks>
		/// <returns>An observable sequence containing the result of pairwise combining the elements of all sources into lists.</returns>
		public static IObservable<IList<TSource>> Zip<TSource>(this IEnumerable<IObservable<TSource>> sources)
		{
			Contract.Requires(sources != null);
			Contract.Ensures(Contract.Result<IObservable<IList<TSource>>>() != null);

			return sources.ToObservable().Zip();
		}
#endif

		/// <summary>
		/// Merges two or more observable sequences into one observable sequence of lists by combining their elements in a pairwise fashion.
		/// </summary>
		/// <typeparam name="TSource">The object that provides notification information.</typeparam>
		/// <param name="sources">An observable sequence containing two or more observable sequences to be merged.</param>
		/// <remarks>
		/// <para>
		/// All of the observable sequences must produce at least one element; otherwise, the generated sequence will be empty.  Furthermore, if an observable 
		/// sequence produces more than one element before each of the other observable sequences have produced their first elements, then all of the elements 
		/// will be enqueued for future pairing.  No elements are discarded.
		/// </para>
		/// <para>
		/// The latest value of an observable sequence is always located in the generated lists at the same index in which that sequence is located in the outer sequence.
		/// For example, the values from the first observable sequence in the outer sequence will always be at index zero (0) in the lists that are generated.
		/// </para>
		/// </remarks>
		/// <returns>An observable sequence containing the result of pairwise combining the elements of all sources into lists.</returns>
		[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Reliability", "CA2000:Dispose objects before losing scope",
			Justification = "The disposable is composited by the subscription.")]
		public static IObservable<IList<TSource>> Zip<TSource>(this IObservable<IObservable<TSource>> sources)
		{
			Contract.Requires(sources != null);
			Contract.Ensures(Contract.Result<IObservable<IList<TSource>>>() != null);

			return Observable.Create<IList<TSource>>(
				observer =>
				{
					var gate = new object();

					var queues = new List<Queue<Notification<TSource>>>();

					bool outerCompleted = false;

					var outerSubscription = new SingleAssignmentDisposable();
					var disposables = new CompositeDisposable(outerSubscription);

					outerSubscription.Disposable = sources.Subscribe(
						source =>
						{
							int index = queues.Count;

							queues.Add(new Queue<Notification<TSource>>());

							var subscription = new SingleAssignmentDisposable();

							disposables.Add(subscription);

							subscription.Disposable = source.Materialize().Subscribe(
								notification =>
								{
									if (notification.Kind == NotificationKind.OnError)
									{
										observer.OnError(notification.Exception);
										return;
									}

									lock (gate)
									{
										bool completeNow = false;
										bool enqueue = !outerCompleted;

										if (outerCompleted)
										{
											if (queues.Count < 2)
											{
												return;
											}

											Contract.Assert(!enqueue);

											var zipped = ZipNext(queues, ref completeNow, notification, index);

											if (zipped == null)
											{
												if (completeNow)
												{
													queues.Clear();
												}
												else
												{
													enqueue = true;
												}
											}
											else
											{
												observer.OnNext(zipped.AsReadOnly());

												Contract.Assert(!enqueue && !completeNow);
											}
										}

										Contract.Assert(!(enqueue && completeNow));

										if (enqueue)
										{
											queues[index].Enqueue(notification);
										}
										else if (completeNow)
										{
											observer.OnCompleted();
										}
									}
								});
						},
						observer.OnError,
						() =>
						{
							bool completeNow;

							lock (gate)
							{
								outerCompleted = true;

								completeNow = queues.Count < 2;

								if (!completeNow)
								{
									while (true)
									{
										var zipped = ZipNext(queues, ref completeNow);

										if (zipped == null)
										{
											if (completeNow)
											{
												queues.Clear();
											}
											break;
										}

										observer.OnNext(zipped.AsReadOnly());
									}
								}
							}

							if (completeNow)
							{
								observer.OnCompleted();
							}
						});

					return disposables;
				});
		}

		private static List<TSource> ZipNext<TSource>(
			IList<Queue<Notification<TSource>>> queues,
			ref bool completeNow,
			Notification<TSource> current = null,
			int currentQueueIndex = -1)
		{
			Contract.Requires(queues != null);
			Contract.Requires((current != null) == (currentQueueIndex > -1));

			bool hasValuesForAllQueues = true;

			for (int i = 0; i < queues.Count; i++)
			{
				Notification<TSource> notification;

				if (i == currentQueueIndex)
				{
					notification = current;
				}
				else
				{
					var queue = queues[i];

					Contract.Assume(queue != null);

					hasValuesForAllQueues &= queue.Count > 0;

					if (queue.Count == 0)
					{
						notification = null;
					}
					else
					{
						notification = queue.Peek();
					}
				}

				if (notification != null && notification.Kind == NotificationKind.OnCompleted)
				{
					completeNow = true;
					return null;
				}
			}

			List<TSource> zipped = null;

			if (hasValuesForAllQueues)
			{
				zipped = new List<TSource>(queues.Count);

				for (int i = 0; i < queues.Count; i++)
				{
					Notification<TSource> notification;

					if (i == currentQueueIndex)
					{
						notification = current;
					}
					else
					{
						var queue = queues[i];

						Contract.Assume(queue != null);

						notification = queue.Dequeue();
					}

					Contract.Assume(notification != null);
					Contract.Assume(notification.Kind == NotificationKind.OnNext);

					zipped.Add(notification.Value);
				}
			}

			return zipped;
		}
	}
}